package com.byron.media.server.handlers;

import com.byron.media.server.MediaServer;
import com.byron.media.server.client.MediaClient;
import com.byron.media.server.model.MediaHeaderObj;
import com.byron.media.server.model.ProxyMediaData;
import com.byron.media.server.utils.JsonUtils;
import com.byron.media.server.utils.MediaType;
import com.byron.media.server.model.DataFactory;
import com.byron.media.server.model.MediaData;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

@Slf4j
public class SessionContainer {

    private Map<String, Set<ISession>> sessionMap = new ConcurrentHashMap<>();

    private Map<String, ReadWriteLock> lockMap = new ConcurrentHashMap<>();

    private Map<String, Queue<MediaData>> iframeQueueMap = new ConcurrentHashMap<>();

    private Set<MediaSession> proxyClientList = new ConcurrentSkipListSet<>();

    private ReadWriteLock proxyClientLock = new ReentrantReadWriteLock();

    private DataFactory dataFactory;

    private MediaClient mediaClient;

    public SessionContainer(DataFactory dataFactory) {
        this.dataFactory = dataFactory;
    }

//    private SessionContainer(){};

//    public void initDataFactory(DataFactory dataFactory){
//        this.dataFactory = dataFactory;
//    }

    /**
     * 添加到组
     */
    public void addToGroup(String device, String group, ISession receiver){
//        String group = sender.getGroupName();
        ReadWriteLock lock = lockMap.computeIfAbsent(group, (k) -> new ReentrantReadWriteLock());
        lock.writeLock().lock();
        try {
            Set<ISession> sessionSet = sessionMap.computeIfAbsent(group, (k) -> new HashSet<>());
            sessionSet.add(receiver);
        } finally {
            lock.writeLock().unlock();
        }
    }


    /**
     * 添加到代理服务器队列
     * @param proxyClient
     */
    public void addToProxyClient(MediaSession proxyClient){
        proxyClientLock.writeLock().lock();
        try {
            proxyClientList.add(proxyClient);
        } finally {
            proxyClientLock.writeLock().unlock();
        }
    }

    /**
     * 移除代理服务器
     * @param proxyClient
     */
    public void removeFromProxyClient(MediaSession proxyClient){
        proxyClientLock.writeLock().lock();
        try {
            proxyClientList.remove(proxyClient);
        } finally {
            proxyClientLock.writeLock().unlock();
        }
    }


    /**
     * 从组移除
     */
    public void removeFromGroup(String device, String group){
        ReadWriteLock lock = lockMap.computeIfAbsent(group, (k) -> new ReentrantReadWriteLock());
        lock.writeLock().lock();
        try {
            Set<ISession> sessionSet = sessionMap.computeIfAbsent(group, (k) -> new HashSet<>());
            sessionSet.removeIf((ses) -> device.equals(ses.getDevice()));
        } finally {
            lock.writeLock().unlock();
        }
    }


    /**
     * 群发消息
     */
    public void sendToGroup(String group, MediaData data){
        ReadWriteLock lock = lockMap.computeIfAbsent(group, (k) -> new ReentrantReadWriteLock());
        lock.readLock().lock();
        try {
            pushLastIframe(group, data);

            Set<ISession> sessionSet = sessionMap.computeIfAbsent(group, (k) -> new HashSet<>());
            for(ISession session : sessionSet){
//                if(session.isProxy()){
//                    session.writeAndFlush(new ProxyMediaData(group, data));
//                    continue;
//                }
                session.write(data);
            }
        } finally {
            lock.readLock().unlock();
        }
    }

    /**
     * 发送最上一次I帧，让客户端恢复界面
     * @param session
     */
    public void sendLastIframe(String group, ISession session) {

        // TODO 对I帧列表添加锁
        ReadWriteLock lock = lockMap.computeIfAbsent(group + "_queue", (k) -> new ReentrantReadWriteLock());
        lock.readLock().lock();
        try {
            Queue<MediaData> iframeQueue = iframeQueueMap.computeIfAbsent(group, (k) -> new ConcurrentLinkedQueue<>());
            if(iframeQueue.isEmpty()){
                return;
            }
            iframeQueue.forEach(session::write);
        } finally {
            lock.readLock().unlock();
        }

    }

    /**
     * TODO 编码器中判断I帧会出问题，导致OOM
     * 将i帧推到队列中
     * @param group
     * @param data
     */
    public void pushLastIframe(String group, MediaData data){
        if(MediaType.isTest(data.getType())){
            return;
        }
        boolean isIframe = MediaType.isIframe(data);
        Queue<MediaData> freeData = null;
        ReadWriteLock lock = lockMap.computeIfAbsent(group + "_queue", (k) -> new ReentrantReadWriteLock());
        lock.writeLock().lock();
        try {
            Queue<MediaData> iframeQueue = iframeQueueMap.computeIfAbsent(group, (k) -> new ConcurrentLinkedQueue<>());
            if(isIframe){
                if(dataFactory != null){
                    freeData = new ConcurrentLinkedQueue<>(iframeQueue);
                }
                iframeQueue.clear();
            }
            iframeQueue.offer(data);
        } finally {
            lock.writeLock().unlock();
        }

        if(freeData != null && dataFactory != null){
            dataFactory.offerDataBatch(freeData);
        }

    }

    /**
     * 清理缓存
     * @param group
     */
    public void cleanIframe(String group) {
        log.info("清空I帧");
        ReadWriteLock lock = lockMap.computeIfAbsent(group + "_queue", (k) -> new ReentrantReadWriteLock());
        lock.writeLock().lock();
        try {
            Queue<MediaData> iframeQueue = iframeQueueMap.computeIfAbsent(group, (k) -> new ConcurrentLinkedQueue<>());
            iframeQueue.clear();
        } finally {
            lock.writeLock().unlock();
        }
    }




    /**
     * 向上层发送流媒体数据
     * @param data
     */
    public void sendToMainServer(Object data) {
        if(mediaClient != null){
            mediaClient.send(data);
        }
    }

    /**
     * 向上层发送命令数据
     * @param header
     */
    public void sendToMainServer(MediaHeaderObj header) {
        if(mediaClient != null){
            mediaClient.send(header);
        }
    }
    /**
     * 向下所有的代理服务器发送指令消息
     * @param header
     */
    public void sendToAllProxyClient(MediaHeaderObj header) {
        proxyClientLock.readLock().lock();
        String json = JsonUtils.toJson(header);
        try {
            proxyClientList.forEach(session -> {
                session.write(json);
            });
        } finally {
            proxyClientLock.readLock().unlock();
        }
    }

    public void setMediaClient(MediaClient mediaClient) {
        this.mediaClient = mediaClient;
    }

    public void broadcastAsRtp(String groupName, MediaData mediaData) {
        if(groupName == null){
            return;
        }
        String[] temp = groupName.split(":");
        if(temp == null || temp.length < 2){
            return;
        }
        String groupAddress = temp[0];
        int port = Integer.parseInt(temp[1]);


    }


//    private static SessionContainer instance;
//    public static SessionContainer getInstance() {
//        if (instance == null) {
//            synchronized (SessionContainer.class) {
//                if (instance == null) {
//                    instance = new SessionContainer(dataFactory, mediaServer);
//                }
//            }
//        }
//        return instance;
//    }

}
